package com.dman.multi.core;

import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;


public class NStream {
	
	public static final int BUFFER_SIZE = 4096;
	public static final int NUM_MODES = 3;
	
	public static enum Mode{
		HEAVY(3), MEDIUM(2), LIGHT(1);
		
		private int number = 0;
		
		private Mode(int i) {
			number = i;
		}

		public int getNumber() {
			return number;
		}
	}
	
	private static Mode mode = Mode.HEAVY;

	private Map<Section, List<SelectionKey>> secKeyLookup = null;
	private transient Selector sockSelector = null;
	private transient Map<Integer, List<Selector>> modeToSelectors;
	private static Map<Integer, Mode> modeMap = new HashMap<Integer, Mode>();
	static{
		modeMap.put(Mode.HEAVY.getNumber(), Mode.HEAVY);
		modeMap.put(Mode.MEDIUM.getNumber(), Mode.MEDIUM);
		modeMap.put(Mode.LIGHT.getNumber(), Mode.LIGHT);
	}

	public static void setMode(Mode mode) {
		NStream.mode = mode;
	}
	
	private int [] modeIter;
	private int [] readIter;
	
	/**
	 * The constructor simply opens a selector to enable sections to register, with
	 * this selector.
	 * 
	 * @throws IOException
	 */
	public NStream() throws IOException{
		modeToSelectors = new HashMap<Integer, List<Selector>>();
		modeIter = new int [NUM_MODES];
		readIter = new int [NUM_MODES];
		for(int i = 1; i <= NUM_MODES; ++i){
			List<Selector> listOfSelector = new ArrayList<Selector>();
			for(int j = 1; j <= i; ++j){
				listOfSelector.add(Selector.open());
			}
			modeToSelectors.put(4 - i, listOfSelector);
			modeIter[i - 1] = 0;
			readIter[i - 1] = 0;
		}
		sockSelector = modeToSelectors.get(Mode.HEAVY.getNumber()).get(0);
		secKeyLookup = new HashMap<Section, List<SelectionKey>>();
	}
	
	/**
	 * This method registers a given section with a selector. The section is 
	 * registered for read operation only. This also attaches the section to the
	 * selection key returned from the selector.
	 * 
	 * @param Section object
	 * @return SelectionKey
	 * @throws IOException
	 */
	public void registerSection(Section sec) throws IOException{
		SocketChannel sockChannel = sec.call();
		Selector sel = null;
		SelectionKey selKey = null;
		List<SelectionKey> listOfSelectionKey = new ArrayList<SelectionKey>();
		for(int i = NUM_MODES - 1; i >= 0; --i){
			sel = modeToSelectors.get(i + 1).get(modeIter[i]);
			++modeIter[i];
			modeIter[i] %= (NUM_MODES - i);
			selKey = sockChannel.register(sel, SelectionKey.OP_READ, sec);
			listOfSelectionKey.add(selKey);
		}
		
		secKeyLookup.put(sec, listOfSelectionKey);
	}
	
	/**
	 * This method performs a select() operation on a selector and returns the set 
	 * of keys which are selected  for read operation.
	 * 
	 * @return Set of SelectionKey
	 * @throws IOException
	 * @throws InterruptedException 
	 */
	public Set<SelectionKey> getReadySockets(long timeout) throws IOException, InterruptedException{
		Set<SelectionKey> setOfSock = new HashSet<SelectionKey>();
		int num = mode.getNumber() - 1;
		List<Selector> selectorList = modeToSelectors.get(num + 1);
		Selector sel = selectorList.get(readIter[num]); 
		++readIter[num];
		readIter[num] %= (NUM_MODES - num);
		if(sel.keys().size() > 0){
			sel.select(timeout);
			setOfSock = sel.selectedKeys();
		}
		return setOfSock;
	}
	
	/**
	 * This method reads a socket into a byte buffer and returns the same. It clears
	 * the buffer before it populates data into it. It also flips the same after
	 * completion of read, so that it can readily be used for writing.
	 * 
	 * @param SelectionKey
	 * @return a ByteBuffer
	 * @throws IOException
	 */
	public ByteBuffer readSocket(SelectionKey selKey) throws EOFException, IOException{
		Section sec = (Section) selKey.attachment();
		ByteBuffer bb = sec.read();
		return bb;
	}
	
	/**
	 * This method re-initializes a section, this is to help out when a socket
	 * is hung, or not responding.
	 * 
	 * @param Section
	 * @throws IOException 
	 */
	public void reinit(Section sec) throws IOException{
		List<SelectionKey> listOfSelectionKey = secKeyLookup.get(sec);
		for(SelectionKey selKey: listOfSelectionKey){
			selKey.cancel();
		}
		sec.close();
		registerSection(sec);
	}
	
	/**
	 * This method closes the sockets if the download is interrupted. This also takes
	 * care of sockets which might already be closed, by the interrupt.
	 * 
	 * @throws IOException
	 */
	public void close() throws IOException {
		Set<SelectionKey> setOfRegisteredSock = sockSelector.keys();
		
		// iterate through selection keys and close the channels
		for(SelectionKey selKey: setOfRegisteredSock){
			SocketChannel sockChannel = (SocketChannel) selKey.channel();
			
			// the condition is important here since
			// it ensure that interrupt has not already closed the socket
			if(sockChannel.isOpen())
				sockChannel.close();
		}
		
		// close selector
		sockSelector.close();
	}
	
	/*public static void main(String[] args) {
		double highestValue = 166;
		double lowestValue = 100;
		double currentValue = 117;
		double averageValue = (highestValue + lowestValue) / 2;
		double numDays = 30;
		System.out.println(averageValue);
		System.out.println((averageValue - currentValue) / currentValue / numDays * 365);
	}*/
}
